package net.bwie.vehicle.dws.function1;

import net.bwie.vehicle.dws.bean1.TripMetrics;
import net.bwie.vehicle.dws.bean1.VehicleData;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
/**
 * 窗口函数类，用于计算车辆行驶统计数据
 * 该函数在时间窗口内处理车辆数据，生成行程指标
 */
public  class ItineraryAnalysisFunction extends ProcessWindowFunction<VehicleData, TripMetrics, String, TimeWindow> {

    private static final Double SPEEDING_THRESHOLD = 80.0;

    /**
     * 处理窗口内的车辆数据，计算各种行程指标
     *
     * @param vin      车辆识别号
     * @param context  窗口上下文
     * @param elements 窗口内的车辆数据集合
     * @param out      结果收集器
     */
    @Override
    public void process(
            String vin,
            Context context,
            Iterable<VehicleData> elements,
            Collector<TripMetrics> out
    ) {
        // 初始化统计变量
        long windowStart = context.window().getStart();
        long windowEnd = context.window().getEnd();
        double totalDistance = 0.0;
        double maxSpeed = 0.0;
        boolean speeding = false;
        int dataPointCount = 0;
        double totalSpeed = 0.0;

        // 记录窗口内第一个和最后一个有效数据点
        VehicleData firstData = null;
        VehicleData lastData = null;

        // 记录前一个数据点用于距离计算
        VehicleData prevData = null;
        double windowDistanceByPosition = 0.0;

        // 遍历窗口内所有数据
        for (VehicleData data : elements) {
            if (firstData == null) {
                firstData = data;
            }
            lastData = data;

            // 更新最高速度
            if (data.getSpeed() > maxSpeed) {
                maxSpeed = data.getSpeed();
            }

            // 检查是否超速
            if (data.getSpeed() > SPEEDING_THRESHOLD) {
                speeding = true;
            }

            // 累计速度用于计算平均速度
            totalSpeed += data.getSpeed();
            dataPointCount++;

            // 计算基于位置的距离
            if (prevData != null) {
                windowDistanceByPosition += calculateDistance(
                        prevData.getLatitude(), prevData.getLongitude(),
                        data.getLatitude(), data.getLongitude());
            }
            prevData = data;
        }

        // 计算窗口内的总里程(优先使用位置计算的距离)
        if (firstData != null && lastData != null) {
            double odometerDistance = lastData.getMileage() - firstData.getMileage();
            totalDistance = windowDistanceByPosition > 0 ? windowDistanceByPosition : odometerDistance;
        }

        // 创建并填充TripMetrics对象
        if (firstData != null && lastData != null) {
            // 创建一个TripMetrics对象来存储行程数据
            TripMetrics metrics = new TripMetrics();

            // 车辆识别号
            metrics.setVin(vin);
            // 行程开始时间
            metrics.setStartTime(windowStart);
            // 行程结束时间
            metrics.setEndTime(windowEnd);
            // 行程总里程
            metrics.setDistance(totalDistance);
            // 计算行程总时长
            metrics.setDuration(windowEnd - windowStart);
            // 计算平均速度，确保不会除以0
            metrics.setAvgSpeed(dataPointCount > 0 ? totalSpeed / dataPointCount : 0.0);
            // 最高时速
            metrics.setMaxSpeed(maxSpeed);
            // 超速状态
            metrics.setSpeeding(speeding);

            // 输出行程数据
            out.collect(metrics);

        }
    }

    /**
     * 使用Haversine公式计算两个地理坐标点之间的距离
     *
     * @param lat1 第一个点的纬度
     * @param lon1 第一个点的经度
     * @param lat2 第二个点的纬度
     * @param lon2 第二个点的经度
     * @return 两点之间的距离（公里）
     */
    private double calculateDistance(double lat1, double lon1, double lat2, double lon2) {
        final int R = 6371; // 地球半径(km)

        double latDistance = Math.toRadians(lat2 - lat1);
        double lonDistance = Math.toRadians(lon2 - lon1);

        double a = Math.sin(latDistance / 2) * Math.sin(latDistance / 2)
                + Math.cos(Math.toRadians(lat1)) * Math.cos(Math.toRadians(lat2))
                * Math.sin(lonDistance / 2) * Math.sin(lonDistance / 2);

        double c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));

        return R * c;
    }
}
